package com.bruce.tool.mq.rocket.util;

import com.bruce.tool.common.exception.BaseRuntimeException;
import com.bruce.tool.mq.rocket.config.RocketConfig;
import com.bruce.tool.mq.rocket.constant.RocketCode;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;

/**
 * 功能 :
 *
 * @author : Bruce(刘正航) 14:26 2019-02-14
 */
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class RocketUtils {

    /**初始化发送者**/
    public static DefaultMQProducer initSender(RocketConfig config){
        DefaultMQProducer producer;
        try {
            producer = new DefaultMQProducer(config.getGroup());
            // 默认消息发送超时时间为3秒(3000)
            producer.setSendMsgTimeout(6000);
            producer.setNamesrvAddr(config.getUrl());
            // 在发送消息前，必须调用 start 方法来启动 Producer，只需调用一次即可
            producer.start();
        } catch (MQClientException e) {
            throw new BaseRuntimeException(RocketCode.MQERROR_SENDER_NULL,"消息发送者初始化失败");
        }
        return producer;
    }

    /**初始化接收者**/
    public static DefaultMQPushConsumer initReceiver(RocketConfig config){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(config.getGroup());
        consumer.setNamesrvAddr(config.getUrl());
        return consumer;
    }
}
